package Sink

import java.sql._
import java.util.UUID

import com.mchange.v2.c3p0.ComboPooledDataSource
import common.MySqlManager
import org.apache.spark.sql.{ForeachWriter, Row}

/**
  *
  * @author Abyss
  * @date 2019/9/3
  * @description
  */
class CustomerJDBCSink(url:String, user:String, pwd:String) extends ForeachWriter[Row]{

  val driver = "com.mysql.jdbc.Driver"
  var connection:Connection=_

  var statement:Statement = _
  var ps: PreparedStatement=_
  var dataSource:ComboPooledDataSource=_
  val sql="replace into gate_customer(id,address,customer_id,org_id) values (?,?,?,?)"
//  val sql="insert into test_gate(address,area) values (?,?)"

  def open(partitionId: Long,version: Long): Boolean = {
//    UUID.randomUUID()

    val manager = MySqlManager.getMDBManager(url, user, pwd)
    dataSource = manager.getDataSource
    connection = manager.getConnection
    ps = connection.prepareStatement(sql)

//
//    Class.forName(driver)
//    connection = DriverManager.getConnection(url, user, pwd)
//    statement = connection.createStatement()

    true
  }
//  "address": "号诺诚1895创意园区10号楼 200072",
//  "area": "静安区",
//  "cableCategory": "01",
//  "capacity": "30",
//  "changeTime": "20190819153214",
//  "channel": "0",
//  "contact": "kevin",
//  "createTime": "20190819153214",
//  "customerCatagory": "电力",
//  "customerId": "30",
//  "customerName": "30",
//  "department": "研发部",
//  "duty": "主管",
//  "email": "kevin@c-goshine.com",
//  "enable": "0",
//  "gender": "男",
//  "idustryCategory": "电力",
//  "mobile": "13800138000",
//  "telephone": "021-52522626"
//  def process(value: Row): Unit = {
//  ps.setObject(1,UUID.randomUUID().toString)
//  ps.setObject(2,value.get(0))
//  ps.setObject(3,value.get(9))
////  ps.setObject(2,value.get(1))
////  ps.setObject(3,value.get(2))
////  ps.setObject(4,value.get(3))
////  ps.setObject(5,value.get(4))
//    val i: Int = ps.executeUpdate()
//  println(" "+value.get(0)+" "+value.get(1)+" "+value.get(2))
//
//  }

  def process(value: Row): Unit = {
      ps.setObject(1,"dsjfksdjfkj")
      ps.setObject(2,"sfsfas")
      ps.setObject(3,"sdfsafs")
      ps.setObject(4,"sdfsafs")
    val i: Int = ps.executeUpdate()
  println( value.get(0)+" "+value.get(1))

  }

//  def process(value: Row): Unit = {
//    val sql = "REPLACE INTO test_gate (`address`,`area`) " +
//      "VALUES (" + '"'+ value.get(0) + '"'+ "," + '"'+ value.get(1)+'"'+  ")"
//
//    println(sql)
//    statement.executeUpdate(sql)
//  }

  def close(errorOrNull: Throwable): Unit = {
//    connection.close()
    dataSource.close()
  }
}
